package com.atguigu.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.annotation.NotSink;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/9 11:13
 */
public class FlinkSinkUtil {
    
    public static SinkFunction<String> getKafkaSink(String topic) {
        
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);
        // TODO
        return new FlinkKafkaProducer<String>(
            "default",
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                @Nullable Long timestamp) {
                    return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);
        
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element,
                                                                @Nullable Long timestamp) {
                    String topic = element.f1.getSink_table();
                    byte[] data = element.f0.toJSONString().getBytes(StandardCharsets.UTF_8);
                    return new ProducerRecord<>(topic, data);
                }
                
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
        
    }
    
    public static void main(String[] args) {
        getClickHouseSink("a", "t", VisitorStats.class);
    }
    
    public static <T> SinkFunction<T> getClickHouseSink(String database, String tableName, Class<T> tClass) {
        
        String driver = "ru.yandex.clickhouse.ClickHouseDriver";
        String url = "jdbc:clickhouse://hadoop162:8123/" + database;
        // insert into user(aaa, bb, cddd)values(?,?,?)
        StringBuilder sql = new StringBuilder();
        
        sql.append("insert into ")
            .append(tableName)
            .append("(")
            .append(AtguiguUtil.getFieldsString(tClass))
            .append(")values(")
            .append(AtguiguUtil.getFieldsString(tClass).replaceAll("[^,]+", "?"))
            .append(")");
        
        //        System.out.println(sql.toString());
        
        return getJdbcSink(sql.toString(), driver, url);
    }
    
    public static <T> SinkFunction<T> getJdbcSink(String sql, String driver, String url) {
        return JdbcSink.sink(
            sql,
            new JdbcStatementBuilder<T>() {
                @Override
                public void accept(PreparedStatement ps,
                                   T t) throws SQLException {
                    System.out.println(t);
                    //insert into t(stt,edt,vc,ch,ar,is_new,uv_ct,pv_ct,sv_ct,uj_ct,dur_sum,ts)values(?,?,?,?,?,?,?,?,?,?,?,?)
                    // 获取类的方式: Class.forName  类名.class  对象.getClass
                    Class<?> tClass = t.getClass();
                    Field[] fields = tClass.getDeclaredFields();
                    try {
                        for (int i = 0, position = 1; i < fields.length; i++) {
                            Field field = fields[i];
    
                            NotSink notSink = field.getAnnotation(NotSink.class);
                            if (notSink == null) {
                                field.setAccessible(true);
                                Object v = field.get(t);
                                if (v == null) {
                                    System.out.println(v);
                                }
                                ps.setObject(position++, v);  // 中间有可能会跳过, 所以要给站位符的索引但是设置计数器
                            }
                        }
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                }
            },
            new JdbcExecutionOptions.Builder()
                .withBatchIntervalMs(1000)
                .withBatchSize(1024)
                .withMaxRetries(3)
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName(driver)
                .withUrl(url)
                .build()
        
        );
        
    }
    
}
